/*
 * Copyright (C) 2008 Universidade Federal de Campina Grande
 *  
 * This file is part of OurGrid. 
 *
 * OurGrid is free software: you can redistribute it and/or modify it under the
 * terms of the GNU Lesser General Public License as published by the Free 
 * Software Foundation, either version 3 of the License, or (at your option) 
 * any later version. 
 * 
 * This program is distributed in the hope that it will be useful, but WITHOUT 
 * ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or 
 * FITNESS FOR A PARTICULAR PURPOSE. See the GNU Lesser General Public License
 * for more details. 
 * 
 * You should have received a copy of the GNU Lesser General Public License 
 * along with this program. If not, see <http://www.gnu.org/licenses/>.
 * 
 */
package org.ourgrid.worker.receiver;

import java.io.File;

import org.ourgrid.broker.actions.ErrorOcurredMessageHandle;
import org.ourgrid.broker.actions.WorkerIsReadyMessageHandle;
import org.ourgrid.common.interfaces.Worker;
import org.ourgrid.common.interfaces.WorkerClient;
import org.ourgrid.common.interfaces.to.GridProcessErrorTypes;
import org.ourgrid.common.interfaces.to.GridProcessHandle;
import org.ourgrid.common.interfaces.to.MessageHandle;
import org.ourgrid.common.interfaces.to.WorkAccounting;
import org.ourgrid.reqtrace.Req;
import org.ourgrid.worker.WorkerConstants;
import org.ourgrid.worker.controller.EnvironmentController;
import org.ourgrid.worker.controller.FileTransferController;
import org.ourgrid.worker.controller.GridProcessError;
import org.ourgrid.worker.controller.UnableToCreatePlaypenException;
import org.ourgrid.worker.controller.UnableToCreateStorageException;
import org.ourgrid.worker.controller.VOMSAuthorizationException;
import org.ourgrid.worker.controller.WorkerController;
import org.ourgrid.worker.dao.EnvironmentDAO;
import org.ourgrid.worker.dao.FileTransferDAO;
import org.ourgrid.worker.dao.WorkAccountingDAO;
import org.ourgrid.worker.dao.WorkerMessageProcessorDAO;
import org.ourgrid.worker.dao.WorkerStatusDAO;
import org.ourgrid.worker.messages.WorkerControllerMessages;

import br.edu.ufcg.lsd.commune.api.FailureNotification;
import br.edu.ufcg.lsd.commune.api.InvokeOnDeploy;
import br.edu.ufcg.lsd.commune.api.MonitoredBy;
import br.edu.ufcg.lsd.commune.api.RecoveryNotification;
import br.edu.ufcg.lsd.commune.container.servicemanager.ServiceManager;
import br.edu.ufcg.lsd.commune.identification.ContainerID;
import br.edu.ufcg.lsd.commune.identification.DeploymentID;
import br.edu.ufcg.lsd.commune.network.certification.CertificationUtils;
import br.edu.ufcg.lsd.commune.processor.filetransfer.IncomingTransferHandle;
import br.edu.ufcg.lsd.commune.processor.filetransfer.OutgoingTransferHandle;
import br.edu.ufcg.lsd.commune.processor.filetransfer.TransferProgress;

/**
 * Worker implementation, called by MyGrid to start work, 
 * The messages are given to WorkerController.
 * 
 * @see org.ourgrid.common.interfaces.control.WorkerControl
 * @see org.ourgrid.worker.controller.WorkerComponentController
 */
@Req("REQ003")
public class WorkerReceiver implements Worker {
	
	private static final long serialVersionUID = 1L;

	private ServiceManager serviceManager;

	@Req("REQ079")
	@InvokeOnDeploy
	public void init(ServiceManager serviceManager) {
		this.serviceManager = serviceManager;
	}
	
	/**
	 * @return the serviceManager
	 */
	protected ServiceManager getServiceManager() {
		return serviceManager;
	}
	
	@Req("REQ079")
	/**
	 * Determines the worker must get prepared for starting its work. For that,
	 * it cannot be working, and both the playpen and the storage directories
	 * must be created. The worker warns its consumer it is ready to work.
	 * 
	 * The worker, however, will only consider this message in case the sender's
	 * public key is the same as the client's one.
	 * 
	 * This message will not be fully completed if either the playpen or the
	 * storage directories are not able to be created.
	 * 
	 * @param workerClient
	 *            the client this worker will be working for.
	 * @param requestID the request id associated to the current work session
	 * @param replicaHandle a handle representing the replica.
	 *            
	 */
	public void startWork(@MonitoredBy(WorkerConstants.WORKER) WorkerClient workerClient,
			long requestID, GridProcessHandle replicaHandle) {
		DeploymentID brokerID = getServiceManager().getStubDeploymentID(workerClient);
		String senderPubKey = getServiceManager().getSenderPublicKey();
		
		WorkerStatusDAO workerStatusDAO = getServiceManager().getDAO(WorkerStatusDAO.class);
		EnvironmentDAO environmentDAO = getServiceManager().getDAO(EnvironmentDAO.class);
		
		if(!senderPubKey.equals(workerStatusDAO.getConsumerPublicKey()) ||
				!senderPubKey.equals(brokerID.getPublicKey())) {
			getServiceManager().getLog().warn(WorkerControllerMessages.getUnknownClientSendsStartWorkMessage(brokerID));
			return;
		}
		
		//Worker is broker-authorized
		if (workerStatusDAO.isAuthorisedForBroker()) {
			
			if(!CertificationUtils.isCertificateValid(serviceManager.getSenderCertPath())) {
				serviceManager.getLog().warn(WorkerControllerMessages.getInvalidCertPathMessage(brokerID));
				
				workerClient.sendMessage(new ErrorOcurredMessageHandle(new GridProcessError(
						new VOMSAuthorizationException(CertificationUtils.getCertSubjectDN(serviceManager.getSenderCertPath())), 
						GridProcessErrorTypes.BROKER_ERROR)));
				
				return;
			}
			
			if(!workerStatusDAO.getUsersDN().isEmpty() && !workerStatusDAO.containsUsersDN(
					CertificationUtils.getCertSubjectDN(serviceManager.getSenderCertPath()))) {
				serviceManager.getLog().warn(WorkerControllerMessages.getBrokerIsNotOnVomsListMessage(brokerID));
				
				workerClient.sendMessage(new ErrorOcurredMessageHandle(new GridProcessError(
						new VOMSAuthorizationException(CertificationUtils.getCertSubjectDN(serviceManager.getSenderCertPath())), 
						GridProcessErrorTypes.BROKER_ERROR)));
				
				return;
			}
			
			if(!CertificationUtils.isCertPathIssuedByCA(serviceManager.getSenderCertPath(), 
					workerStatusDAO.getCaCertificates())) {
				serviceManager.getLog().warn(WorkerControllerMessages.getNonIssuedCertPathMessage(brokerID));
				
				workerClient.sendMessage(new ErrorOcurredMessageHandle(new GridProcessError(
						new VOMSAuthorizationException(CertificationUtils.getCertSubjectDN(serviceManager.getSenderCertPath())),
						GridProcessErrorTypes.BROKER_ERROR)));
				
				return;
			}
			
		}
		
		if (workerStatusDAO.isAllocated()) {
			WorkerController.getInstance().cleanWorker(getServiceManager());
		}
		
		try {
			EnvironmentController.getInstance().mountPlaypen(getServiceManager());
		} catch (UnableToCreatePlaypenException e) {
			getServiceManager().getLog().error(WorkerControllerMessages.getPlaypenCreationErrorMessage(brokerID, e.getPlaypenPath()));
			workerClient.sendMessage(new ErrorOcurredMessageHandle(new GridProcessError(e, GridProcessErrorTypes.IO_ERROR)));
			environmentDAO.resetEnvVariables();
			return;
		}
		
		try {
			EnvironmentController.getInstance().mountStorage(
					workerStatusDAO.getConsumerPublicKey(), getServiceManager());
		} catch (UnableToCreateStorageException e) {
			getServiceManager().getLog().error(WorkerControllerMessages.getStorageCreationErrorMessage(brokerID, e.getStoragePath()));
			workerClient.sendMessage(new ErrorOcurredMessageHandle(new GridProcessError(e, GridProcessErrorTypes.IO_ERROR)));
			environmentDAO.resetEnvVariables();
			return;
		}
		
		if(workerStatusDAO.isAllocatedForRemotePeer()) {
			getServiceManager().getDAO(WorkAccountingDAO.class).
			setActualWorkAccounting(new WorkAccounting(workerStatusDAO.getRemotePeerPublicKey()));
		}
				
		workerStatusDAO.setConsumer(workerClient);
		getServiceManager().getLog().info(WorkerControllerMessages.getSuccessfulStartWorkMessage(brokerID));
		workerClient.sendMessage(new WorkerIsReadyMessageHandle());
	}
	
	public void sendMessage(MessageHandle handle) {
		getWorkerMessageProcessorDAO().getMessageProcessor(
				handle.getActionName()).process(handle, getServiceManager());
	}
	
	@Req("REQ080")
	public void transferRequestReceived(IncomingTransferHandle handle){
		ContainerID senderID = handle.getSenderID();
		
		WorkerClient consumer = getWorkerStatusDAO().getConsumer();
		
		if (consumer == null) {
			FileTransferController.getInstance().rejectTransferRequest(handle, getServiceManager());
			getServiceManager().getLog().warn(WorkerControllerMessages.
					getUnknownClientRequestsToTransferFileMessage(handle.getDestinationFileName(),
							handle.getFileSize(), handle.getId(), handle.getSenderID().toString()));
			return; //TODO log
		}
		
		DeploymentID consumerID = serviceManager.getStubDeploymentID(consumer);
		
		if (consumerID.getContainerID().equals(senderID)) {
			handle.setSenderID(consumerID.getContainerID());
		
		} else {
			getServiceManager().getLog().warn(WorkerControllerMessages.
					getUnknownClientRequestsToTransferFileMessage(handle.getDestinationFileName(),
							handle.getFileSize(), handle.getId(), handle.getSenderID().toString()));
			FileTransferController.getInstance().rejectTransferRequest(handle, getServiceManager());
			return;
		}
			
		String senderPublicKey = consumerID.getPublicKey();
		
		if(!getWorkerStatusDAO().hasConsumer()) {
			getServiceManager().getLog().warn(WorkerControllerMessages.
					getClientTriesToTransferFileOnUnstartedWorkerMessage(senderPublicKey));
			FileTransferController.getInstance().rejectTransferRequest(handle, getServiceManager());
			return;
		}
		
		if(getWorkerStatusDAO().isErrorState()) {
			getServiceManager().getLog().warn(WorkerControllerMessages.
					getClientRequestsToTransferFileOnWorkerWithErrorMessage(handle.getDestinationFileName(),
							handle.getFileSize(), handle.getId(), senderPublicKey));
			return;
		}
		
		FileTransferController.getInstance().acceptTransferRequest(handle, getServiceManager());
	}
	
	@Req("REQ080")
	public void incomingTransferFailed(IncomingTransferHandle handle, Exception failCause, long amountWritten) {
		
		String senderPublicKey = handle.getSenderID().getPublicKey();

		if(!getWorkerStatusDAO().getConsumerPublicKey().equals(senderPublicKey)) {
			getServiceManager().getLog().warn(WorkerControllerMessages.
					getUnknownClientSendsIncomingTransferFailedMessage(senderPublicKey));
		    return;
		}

		File incomingFileFailed = getFileTransferDAO().getIncomingFile(handle);
		
		if(incomingFileFailed == null) {
			getServiceManager().getLog().warn(WorkerControllerMessages.
					getIncomingTranferFailedWithUnknownHandleMessage(handle.getId(), senderPublicKey));
			return;
		}
		
		if(getWorkerStatusDAO().isErrorState()) {
			getServiceManager().getLog().warn(WorkerControllerMessages.
					getWorkerWithErrorReceivesAnIncomingTransferFailedMessage(handle.getId(), amountWritten, senderPublicKey));
			return;
		}
		
		FileTransferController.getInstance().incomingTransferFailed(handle, failCause, amountWritten, getServiceManager());
	}

	@Req("REQ080")
	public void incomingTransferCompleted(IncomingTransferHandle handle, long amountWritten) {
		
		String senderPublicKey = handle.getSenderID().getPublicKey();

		if(!getWorkerStatusDAO().getConsumerPublicKey().equals(senderPublicKey)) {
			getServiceManager().getLog().warn(WorkerControllerMessages.
					getUnknownClientSendsIncomingTransferCompletedMessage(senderPublicKey));
		    return;
		}
		
		if(getFileTransferDAO().getIncomingFile(handle) == null) {
			getServiceManager().getLog().warn(WorkerControllerMessages.
					getIncomingTranferCompletedWithUnknownHandleMessage(handle.getId(), senderPublicKey));
			return;
		}

		if(getWorkerStatusDAO().isErrorState()) {
			getServiceManager().getLog().warn(WorkerControllerMessages.
					getWorkerWithErrorReceivesAnIncomingTransferCompletedMessage(handle.getId(), amountWritten, senderPublicKey));
			return;
		}
		
		FileTransferController.getInstance().incomingTransferCompleted(handle, amountWritten, getServiceManager());
	}
	
	@Req("REQ0123")
	public void updateTransferProgress(TransferProgress transferProgress) {
		transferProgress.getNewStatus();
		
		WorkAccounting accounting = getServiceManager().getDAO(WorkAccountingDAO.class).getActualWorkAccounting();
		
		if (accounting == null) {
			return;
		}
		
		getServiceManager().getDAO(WorkAccountingDAO.class).getActualWorkAccounting().incDataTransfered(transferProgress.getAmountWritten());
	}


	@Req("REQ081")
	public void transferRejected(OutgoingTransferHandle handle) {
		
		File fileRejected = getFileTransferDAO().getUploadingFile(handle);
		String senderPublicKey = getServiceManager().getSenderPublicKey();
		
		if(fileRejected == null) {
			getServiceManager().getLog().warn(WorkerControllerMessages.
					getWorkerReceivesAFileTransferRejectedWithUnknownHandleMessage(handle.getId(), senderPublicKey));
			return;
		}
		
		String filePath = fileRejected.getAbsolutePath();
		
		if(getWorkerStatusDAO().isErrorState()) {
			getServiceManager().getLog().warn(WorkerControllerMessages.
					getWorkerWithErrorReceivesAFileTransferRejectedMessage(filePath, handle.getId(), senderPublicKey));
			return;
		}
		
		FileTransferController.getInstance().fileRejected(handle, getServiceManager());
	}
	
	@Req("REQ081")
	public void outgoingTransferCancelled(OutgoingTransferHandle handle, long amountUploaded) {
		
		File fileCancelled = getFileTransferDAO().getUploadingFile(handle);
		String senderPublicKey = getServiceManager().getSenderPublicKey();
		
		if(fileCancelled == null) {
			getServiceManager().getLog().warn(WorkerControllerMessages.
					getWorkerReceivesAnOutgoingFileTransferCancelledWithUnknownHandleMessage(handle, amountUploaded, senderPublicKey));
			return;
		}
		
		String filePath = fileCancelled.getAbsolutePath();
		
		if(getWorkerStatusDAO().isErrorState()) {
			getServiceManager().getLog().warn(WorkerControllerMessages.
					getWorkerWithErrorReceivesAnOutgoingFileTransferCancelledMessage(filePath, handle.getId(), amountUploaded, senderPublicKey));
			return;
		}
		
		FileTransferController.getInstance().outgoingTransferCancelled(handle, amountUploaded, getServiceManager());
	}
	
	@Req("REQ081")
	public void outgoingTransferFailed(OutgoingTransferHandle handle, Exception exception, long amountUploaded) {
		
		File fileRejected = getFileTransferDAO().getUploadingFile(handle);
		String senderPublicKey = getServiceManager().getSenderPublicKey();
		
		if(fileRejected == null) {
			getServiceManager().getLog().warn(WorkerControllerMessages.
					getWorkerReceivesAnOutgoingFileTransferFailedWithUnknownHandleMessage(handle.getId(), amountUploaded, senderPublicKey));
			return;
		}
		
		String filePath = fileRejected.getAbsolutePath();
		
		if(getWorkerStatusDAO().isErrorState()) {
			getServiceManager().getLog().warn(WorkerControllerMessages.
					getWorkerWithErrorReceivesAnOutgoingFileTransferFailedMessage(filePath, handle.getId(), amountUploaded, senderPublicKey));
			return;
		}
		
		FileTransferController.getInstance().outgoingTransferFailed(handle, exception, amountUploaded, getServiceManager());
	}

	@Req("REQ081")
	public void outgoingTransferCompleted(OutgoingTransferHandle handle, long amountUploaded) {
		
		File fileCompleted = getFileTransferDAO().getUploadingFile(handle);
		String senderPublicKey = getServiceManager().getSenderPublicKey();
		
		if(fileCompleted == null) {
			getServiceManager().getLog().warn(WorkerControllerMessages.
					getWorkerReceivesAnOutgoingFileTransferCompletedWithUnknownHandleMessage(handle.getId(), amountUploaded, senderPublicKey));
			return;
		}
		
		String filePath = fileCompleted.getAbsolutePath();
		
		if(getWorkerStatusDAO().isErrorState()) {
			getServiceManager().getLog().warn(WorkerControllerMessages.
					getWorkerWithErrorReceivesAnOutgoingFileTransferCompletedMessage(filePath, handle.getId(), amountUploaded, senderPublicKey));
			return;
		}
	
		FileTransferController.getInstance().outgoingTransferCompleted(handle, amountUploaded, getServiceManager());
	}
	
	private WorkerStatusDAO getWorkerStatusDAO() {
		return getServiceManager().getDAO(WorkerStatusDAO.class);
	}
	
	private WorkerMessageProcessorDAO getWorkerMessageProcessorDAO() {
		return getServiceManager().getDAO(WorkerMessageProcessorDAO.class);
	}

	private FileTransferDAO getFileTransferDAO() {
		return getServiceManager().getDAO(FileTransferDAO.class);
	}

	public void shutdown(boolean force) {
		// TODO Auto-generated method stub
	}
	
	@RecoveryNotification
	public void workerClientIsUp(WorkerClient workerClient) {
		getWorkerStatusDAO().setConsumer(workerClient);
	}
	
	@FailureNotification
	public void workerClientIsDown(WorkerClient workerClient) {
		getWorkerStatusDAO().setConsumer(null);
//		serviceManager.release(workerClient);
	}
	
}